package com.aispeech.task.executor.service.impl;

import com.aispeech.task.common.data.constant.Progress;
import com.aispeech.task.common.data.domain.Task;
import com.aispeech.task.common.data.enums.APICode;
import com.aispeech.task.common.exceptions.TaskException;
import com.aispeech.task.common.handler.TaskHandler;
import com.aispeech.task.common.handler.TaskManager;
import com.aispeech.task.common.utils.TaskUtils;
import com.aispeech.task.executor.config.TaskConfig;
import com.aispeech.task.executor.service.AsyncTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

import static com.aispeech.task.common.config.RedisConfig.*;

@Service
@Slf4j
public class AsyncTaskServiceImpl implements AsyncTaskService, TaskManager {
    @Autowired
    @Qualifier("hashTaskOperations")
    private HashOperations<String, String, Task> hashTaskOperations;

    @Autowired
    private StreamOperations<String, String, String> opStream;

    @Autowired
    TaskConfig taskConfig;

    @Autowired
    private TaskHandler taskHandler;

    private volatile boolean running = true;

    @Async("asyncServiceExecutor")
    @Override
    public void blockingReceiveInThreadpool(int i) {
        log.info("begin AsyncServiceImpl.execute, index: {}", i);
        String nodeName = System.getenv().getOrDefault("NODENAME", "dev");

        taskHandler.initialize(this, taskConfig.getEventThreads());

        while (running) {
            try {
                List<MapRecord<String, String, String>> records = opStream.read(
                        Consumer.from(
                                STREAM_GROUP,
                                String.format("%s-%s", nodeName, Thread.currentThread().getName())
                        ),
                        StreamOffset.create(
                                STREAM_NAME,
                                ReadOffset.lastConsumed()
                        )
                );

                records.forEach(entries -> {
                    log.info("[ read ] record id: {}", entries.getId());
                    entries.forEach(this::received);

                    opStream.acknowledge(STREAM_NAME, STREAM_GROUP, entries.getId());
                });
            } catch (Exception ex) {
                log.warn("read redis failed.", ex);
            }

            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                log.error("", e);
            }
        }
    }

    @Override
    public void shutdown() {
        running = false;
        log.info("Shutting down Async Service...");
    }

    private void received(Map.Entry<String, String> entry) {
        log.info("[ received ] : key: {}, value: {}", entry.getKey(), entry.getValue());

        if ("taskId".equals(entry.getKey())) {
            Task task = hashTaskOperations.get(HASH_TASK_KEY, entry.getValue());

            if (task != null) {
                try {
                    taskHandler.execute(task);
                } catch (TaskException e) {
                    log.error("execute err.", e);
                }
            }
        }
    }

    private void save(Task task) {
        task.setTsUpdate(System.currentTimeMillis());
        hashTaskOperations.put(HASH_TASK_KEY, TaskUtils.field(task.getId(), task.getProductId()), task);
    }

    @Override
    public void updateProgress(Task task, int progress) throws TaskException {
        task.setProgress(progress);
        save(task);

        log.info("[ updateProgress ] task: {}, progress: {}", TaskUtils.field(task.getId(), task.getProductId()), progress);
    }

    @Override
    public void finishTask(Task task, List<Map> result) throws TaskException {
        task.getMetrics().put("ts_finished", System.currentTimeMillis());
        task.setProgress(Progress.FINISHED);
        task.setFinished(true);
        task.setResult(result);

        save(task);

        log.info("[ finishTask ] task: {}, result: {}", TaskUtils.field(task.getId(), task.getProductId()), result);
    }

    @Override
    public void breakTask(Task task, APICode code) throws TaskException {
        breakTask(task, code.getCode(), code.getMessage());
    }

    @Override
    public void breakTask(Task task, int code, String message) throws TaskException {
        task.updateError(code, message);

        task.getMetrics().put("ts_finished", System.currentTimeMillis());
        task.setFinished(true);

        save(task);

        log.error("[ breakTask ] task: {}, error: {} - {}", TaskUtils.field(task.getId(), task.getProductId()), code, message);
    }
}
